#include "stdafx.h"

#include <map>
#include <vector>


#include <unistd.h>

#include "mqtt_cli.h"

#include "../../../include/fore/rtdb_fun.h"

#include "../../../include/common/threadLock.h"

#include "../../../include/common/collector.h"
#include "../../../include/common/commFun.h"


#include "../../../include/rapidjson/rapidjson.h" 
#include "../../../include/rapidjson/document.h" 
#include "../../../include/rapidjson/writer.h" 
#include "../../../include/rapidjson/stringbuffer.h" 
#include "../../../include/rapidjson/prettywriter.h" 
#include "../../../include/rapidjson/filereadstream.h"
#include "../../../include/rapidjson/filewritestream.h"
  

using namespace std; 
using namespace rapidjson;


volatile  int   on_connect_ok = 0;
  
    
int Pub_msgarrvd(void* context, char* topicName, int topicLen, MQTTAsync_message* m); 
thread_return_type  pubtopic_loop_check(void* context);
 
thread_return_type  process_pubMsg(void* n);
  
CThreadLock		 g_threadLock; 

CMQTTAsyncPubCli::CMQTTAsyncPubCli()
{  
	//g_repubtopic = false; 
	m_client = NULL;
}

CMQTTAsyncPubCli::~CMQTTAsyncPubCli()
{
	disconnect();
}
  
 

void pub_connlost(void *context, char *cause)
{
	base::my_printf(  "\n pub Connection lost \n" );  
	 
}


void onPubDisconnect(void* context, MQTTAsync_successData* response)
{
	base::my_printf("Successful disconnection\n"); 
}


void onPubConnectFailure(void* context, MQTTAsync_failureData* response)
{
	base::my_printf("Connect failed, rc %d\n", response ? response->code : 0); 
} 


void onPubConnect(void* context, MQTTAsync_successData* response)
{
	on_connect_ok = 1;  
}



void onPubSend(void* context, MQTTAsync_successData* response)
{
	 
}

bool CMQTTAsyncPubCli::re_connect_mqtt()
{  

	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	  
	MQTTAsync_create(&m_client, m_mqtt_svr_addr.c_str(), PUB_CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
	 
	MQTTAsync_setCallbacks(m_client, this,  pub_connlost, Pub_msgarrvd, NULL);
	
	int rc;

	conn_opts.keepAliveInterval = 20;
	conn_opts.cleansession = 1;
	conn_opts.onSuccess = onPubConnect;
	conn_opts.onFailure = onPubConnectFailure;
	conn_opts.context = m_client;
	if ((rc = MQTTAsync_connect(m_client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		base::my_printf("Failed to start connect, return code %d\n", rc);
		return false;
	} 
	 
	int   wait_seconds = 6; 
	while ( !on_connect_ok )
	{ 
		if(wait_seconds-- <= 0) break;
		
		base::MySleep(1000L); 
	}
	 
	if( !on_connect_ok ) {
		base::my_printf("MQTTAsync_connect Failed \n");
		return false;
	}

	base::my_printf("pub_topic: connect success \n" );
 
	create_devTypePubtopic_thread();
	 
	base::MySleep(1000*5); 
 	base::Thread_start(pubtopic_loop_check, (void*)this); 

	return true;
}
 

bool CMQTTAsyncPubCli::ReStartPubThreasd( )
{  
	CloseThreadHandle();	  
	base::MySleep(1000L); 

	create_devTypePubtopic_thread();
	 

  	return true;
	 
}

bool CMQTTAsyncPubCli::connect_mqtt(char * mqtt_svr_addr)
{ 
	if( strlen( mqtt_svr_addr ) == 0 ) return false;

	m_mqtt_svr_addr = mqtt_svr_addr;
	on_connect_ok = 0;

	disconnect();

	re_connect_mqtt(); 

	return true;
}

bool test_thread(thread_type handle)
{ 
        //������߳�thread_t ��Чʱ��  pthread_kill(handle, 0);����ֶδ���
	int kill_rc = pthread_kill(handle, 0);
	//base::my_printf("2222 \n");
	if( kill_rc == ESRCH ) {
		//base::my_printf("the specified thread did not exists or already quit \n");
		return false;
	}
	else if( kill_rc == EINVAL ) {
		//base::my_printf("signal is invalid \n");
		return false;
	}
	else {
		//base::my_printf("rc = %d \n", kill_rc);
		return true;
	}
}

void CMQTTAsyncPubCli::CloseThreadHandle(char * devType )
{
 
	if( devType )
	{
		base::my_printf("---------------------------------begin close pubtopic thread : devType=%s \n", devType );

		g_threadLock.Lock();
		 
		std::map<std::string, THREAD_STATUS>::iterator it1 = m_aryDevTypeThreadHandle.find( devType );
		if(it1 != m_aryDevTypeThreadHandle.end()) 
		{
			THREAD_STATUS  thread_status =  it1->second;

			if( thread_status.status ) { 
				thread_status.status = 0;
				m_aryDevTypeThreadHandle[devType] = thread_status; 
			}
			else {
				
			}
		
			base::my_printf("end close pubtopic thread : topic=%s \n", devType ); 
		}
		 
		g_threadLock.UnLock();

	}
	else
	{
		base::my_printf("-----------------------------------close devtype thread2 begin size=%d...\n", m_aryDevTypeThreadHandle.size() );
			
		g_threadLock.Lock();
		 
		for (std::map<string, THREAD_STATUS>::iterator it = m_aryDevTypeThreadHandle.begin(); it != m_aryDevTypeThreadHandle.end();  it++) 
		{   
			string   devType = (*it).first; 
			THREAD_STATUS    threadstatus =  (THREAD_STATUS)(*it).second;  
			 
			if( threadstatus.status ) { 
				pthread_cancel( threadstatus.handle );
				base::MySleep(100); 
			}
		
			base::my_printf("---------------------------begin close devtype thread : devtype=%s \n", devType.c_str() );
		}
		g_threadLock.UnLock();
	
		base::my_printf("close devtype thread end \n" );
  
		m_aryDevTypeThreadHandle.clear();
	} 
}


thread_return_type  pubtopic_loop_check(void* context)
{
	CMQTTAsyncPubCli  * pPubCli = (CMQTTAsyncPubCli*)context;
	base::my_printf("---------------------------------create check_devtype_loop---------------------------------- \n" );
 
	 
	while(1)
	{
		std::vector<FORE_DEV_TYPE>   aryDevType;
		TDbFun::GetDevType(aryDevType); 
		
		base::my_printf("---------------------------------create check_devtype_loop size=%d \n", pPubCli->m_aryDevTypeThreadHandle.size() );

		g_threadLock.Lock(); 
 
   
		for (std::map<string, THREAD_STATUS>::iterator it = pPubCli->m_aryDevTypeThreadHandle.begin(); it != pPubCli->m_aryDevTypeThreadHandle.end(); ) 
		{   
			string   devDevType 		= (*it).first; 
			THREAD_STATUS    threadstatus 	=  (THREAD_STATUS)(*it).second;  
			 
			if( threadstatus.status == 0 ) {   
			}
			else {  

				bool bfind = false;
				for( int i=0; i<aryDevType.size(); i++ )
				{ 
					if( strcmp(devDevType.c_str(), aryDevType[i].dev_type) == 0 ) {
						bfind = true;
						break;
					}
				}
			
				if( bfind )  { 
					it++;
					continue;
				}  
				 
				pthread_cancel( threadstatus.handle );
				base::MySleep(100); 
			} 
			pPubCli->m_aryDevTypeThreadHandle.erase( it++ ); 
			base::my_printf(" remove devDevType=%s \n", devDevType.c_str() );
		}

		base::my_printf(" ===============check_devtype_loop size=%d \n", pPubCli->m_aryDevTypeThreadHandle.size() );

		for( int i=0; i<aryDevType.size(); i++ )
		{
			//if( aryDevType[i].protocolType == NULL || aryDevType[i].manufacturerId == NULL ) continue;
			if( aryDevType[i].profilename == NULL  ) continue;

			std::map<std::string, THREAD_STATUS>::iterator it1 = pPubCli->m_aryDevTypeThreadHandle.find( aryDevType[i].dev_type );
			if(it1 == pPubCli->m_aryDevTypeThreadHandle.end()) {  //���豸����
				base::my_printf("---------------------------------rebuild create   thread : devDevType=%s \n", aryDevType[i].dev_type );
				pPubCli->create_devTypePubtopic_thread( aryDevType[i].dev_type );
			} 
		} 

		base::my_printf(" *******************check_devtype_loop size=%d \n", pPubCli->m_aryDevTypeThreadHandle.size() );

		g_threadLock.UnLock();

		base::MySleep(1000 * 20);
	}  
	return NULL;
}



void CMQTTAsyncPubCli::disconnect()
{ 
	base::my_printf("pub disconnect \n");

	CloseThreadHandle();
	
	if( MQTTAsync_isConnected( m_client ) ) 
	{ 
		MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
		int rc;  
		 
		opts.onSuccess = onPubDisconnect;
		opts.context = m_client;

		if ((rc = MQTTAsync_disconnect(m_client, &opts)) != MQTTASYNC_SUCCESS)
		{
			base::my_printf("Failed to MQTTAsync_disconnect, return code %d\n", rc);
			exit(EXIT_FAILURE);
		} 
	
		MQTTAsync_destroy(&m_client);
	}
}	 
	  

void CMQTTAsyncPubCli::create_devTypePubtopic_thread(char * devType)
{   
	if( devType == NULL )
	{
		base::my_printf(" create_devTypePubtopic_thread:  load all pubtopic \n"  );
  
		////��ʵʱ�⣬ �������и߼�app���ĵ�����	
		std::vector<FORE_DEV_TYPE>   aryDevType;
		TDbFun::GetDevType( aryDevType ); 
 
		base::my_printf( " devType.size=%d  \n", aryDevType.size() );

		 
		int ii = 0;
		for (int i=0; i<aryDevType.size(); i++) {  
			
			FORE_DEV_TYPE & devTypeInfo = aryDevType[i];

			if( devTypeInfo.protocolType == NULL || devTypeInfo.manufacturerId == NULL ) continue;

//			if( strcmp(devTypeInfo.dev_type, "JC" ) != 0 ) continue;
			  
			struct pub_thread_parms * parms = new struct pub_thread_parms; // { &client, QOS, ii++, ary, (char*)strTopic.c_str() };
			parms->c = m_client;
			parms->qos = QOS;
			parms->index = ii++;
			parms->devTypeInfo = devTypeInfo;
			parms->pPubCli = this;
			  
			pthread_t threadId = base::Thread_start(process_pubMsg, (void*)parms);  
			THREAD_STATUS  threadstatus;
			threadstatus.handle = threadId;
			threadstatus.status = 1;
 
			m_aryDevTypeThreadHandle.insert(pair<std::string, THREAD_STATUS>( devTypeInfo.dev_type, threadstatus));
	
  
			base::MySleep(500L);
		}  

	}
	else 
	{
		base::my_printf(" create_devTypePubtopics_thread:  create devType: %s \n", devType  );
 
		 
		FORE_DEV_TYPE * pRec = (FORE_DEV_TYPE*)FindRec_stringkey(FORE_DEV_TYPE_NO, devType);
		if( pRec == NULL) return  ;

		if( pRec->protocolType == NULL || pRec->manufacturerId == NULL ) return;
		 
//		if( strcmp(pRec->dev_type, "JC" ) != 0 ) return;

		struct pub_thread_parms * parms = new struct pub_thread_parms; 
		parms->c = m_client;
		parms->qos = QOS;
		parms->index = m_aryDevTypeThreadHandle.size();
		memcpy(&parms->devTypeInfo, pRec, sizeof(FORE_DEV_TYPE));
		parms->pPubCli = this; 
 
		thread_type threadid = base::Thread_start(process_pubMsg, (void*)parms);   

		THREAD_STATUS  threadstatus;
		threadstatus.handle = threadid;
		threadstatus.status = 1;

		m_aryDevTypeThreadHandle.insert(pair<std::string, THREAD_STATUS>( devType, threadstatus)); 
  

	}
	 
}

bool _send_event_data(void* context, int qos, FORE_DEV_DICT * pDevInfo, int updateTime, int iEventVal)
{
	if( context == NULL  ) return false;

	MQTTAsync  client = (MQTTAsync)context;
	
	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	MQTTAsync_message pubmsg = MQTTAsync_message_initializer;

	int pos = 0; 
	int SEND_MAX_REC = 50;  
	int idx = 0;

	char  topic[128] = {0};
	sprintf(topic, "%s/terminal/dataReport/%s/%s/%s/%s/%s", "dataCenterApp", pDevInfo->manufacturerId, pDevInfo->manufacturerName,
					pDevInfo->deviceType, pDevInfo->model,  pDevInfo->dev_nodeID );

	rapidjson::StringBuffer s;
	rapidjson::Writer<rapidjson::StringBuffer> writer(s);
 	
	char  tmp[32] = {0}; 
	int  token = 10000; 
	token = getpid();


	writer.StartObject();
	writer.Key("token");
	sprintf(tmp, "%d", token );
	writer.String(tmp);
	writer.Key("timestamp"); 
	writer.String( base::toTimeString(updateTime).c_str() );
	writer.Key("body");
	writer.StartObject();
 
	writer.Key("serviceId"); 
	writer.String("Events");  

	writer.Key("serviceProperties"); 
	writer.StartObject(); 
 
 
	int preVal = iEventVal;
	if( iEventVal == 0 ) {
		preVal = 1;
	}
	else if( iEventVal == 1 ) {
		preVal = 0;
	}

	writer.Key("PwrOffAlmPreVal");   
	sprintf(tmp, "%d", preVal );
	writer.String(tmp); 

	writer.Key("PwrOffAlmCurVal");   
	sprintf(tmp, "%d", iEventVal );
	writer.String(tmp); 

	writer.Key("EventTime");   
	writer.String( base::toTimeString(updateTime).c_str() );
	 
	writer.EndObject();
	writer.EndObject();
	writer.EndObject();  
		 
	pubmsg.payload = (char*)s.GetString();
	pubmsg.payloadlen = s.GetSize() ;  
	pubmsg.retained = 0;
	pubmsg.qos = qos;
				
	opts.onSuccess =  onPubSend;
	opts.context = client;  

	base::my_printf( "send payload: %s\n", s.GetString());

 	int rc = MQTTAsync_send(client, topic, pubmsg.payloadlen, pubmsg.payload,pubmsg.qos, pubmsg.retained, &opts);
	if ( rc  != MQTTASYNC_SUCCESS )
	{  
		base::my_printf( "Failed to  sendMessage  , return code %d\n", rc);
		return false;
	} 

	return true;
}

void _remove_invalid_event_rec()
{ 
	tHandle handle = OpenLib(FORE_EVENT_POINT_NO);
	if( !IsHandleValid(handle) ) {
		printf("FORE_EVENT_POINT_NO is InValid \n");
		return ;
	}

	int  curTime = time(NULL);
	FORE_EVENT_POINT   rec;
	bool ret = ReadFirst(handle, (char*)&rec, sizeof(FORE_EVENT_POINT));

	while( ret )
	{   
		if( (curTime - 60*60*24*10) >= rec.event_time ) {   //dele 10days 
			Delete_key(FORE_EVENT_POINT_NO, rec.event_time);
		} 
		 
		ret = ReadNext(handle, (char*)&rec, sizeof(FORE_EVENT_POINT));
	}
	
	CloseLib( handle );

}

void _Put_Event(FORE_DEV_DICT * pDevInfo, int eventTime, int iEventVal)
{ 
	if( pDevInfo == NULL || eventTime == 0  ) return;

	int rec_count =  GetRecCount(FORE_EVENT_POINT_NO) ;
	if( rec_count >= (FORE_EVENT_POINT_COUNT-200) ) {
		_remove_invalid_event_rec(); 
	}

	FORE_EVENT_POINT  rec;
	rec.event_time = eventTime;
	strcpy(rec.dev_nodeID, pDevInfo->dev_nodeID);
	strcpy(rec.name, "PwrOffAlm");
	strcpy(rec.dev_type, pDevInfo->deviceType);
	strcpy(rec.appName, "msgSvr");

	int preVal = iEventVal;
	if( iEventVal == 0 ) {
		rec.pre_status = 1;
	}
	else if( iEventVal == 1 ) {
		rec.pre_status = 0;
	}
	rec.cur_status = iEventVal; 

	UpdateRec_Ex(FORE_EVENT_POINT_NO, eventTime , (char*)&rec, sizeof(FORE_EVENT_POINT)  ) ; 
}



bool _send_sub_data(void* context, int qos, FORE_DEV_DICT * pDevInfo, char* pServiceID, std::vector<RT_DATA> & rtData, unsigned int load_rt_type )
{ 
	if( context == NULL || pDevInfo == NULL ) return false;

	if( rtData.size() <= 0 ) return false;

	MQTTAsync  client = (MQTTAsync)context;
	
	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
	   
	 
	int pos = 0;
	int COUNT = rtData.size();
	int SEND_MAX_REC = 50;  
	int idx = 0;

	char  topic[128] = {0};
	sprintf(topic, "%s/terminal/dataReport/%s/%s/%s/%s/%s", "dataCenterApp", pDevInfo->manufacturerId, pDevInfo->manufacturerName,
					pDevInfo->deviceType, pDevInfo->model,  pDevInfo->dev_nodeID );


	
	int  token = 10000;
 
	token = getpid();

	int updateTime = 0;

	for( int i=0; i<rtData.size(); i++ )
	{
		RT_DATA & rt = rtData[i]; 

		if( rt.update_time > 0 ) {
			updateTime = rt.update_time;
			break;
		}
	}
 
	if( updateTime <= 0 ) return false;
 
 	 
	rapidjson::StringBuffer s;
	rapidjson::Writer<rapidjson::StringBuffer> writer(s);
 	
	char  tmp[32] = {0}; 

	writer.StartObject();
	writer.Key("token");
	sprintf(tmp, "%d", token );
	writer.String(tmp);
	writer.Key("timestamp"); 
	writer.String( base::toTimeString(updateTime).c_str() );
	writer.Key("body");
	writer.StartObject();

	 
	writer.Key("serviceId");
	sprintf(tmp, "%s", pServiceID );
	writer.String(tmp);  

	writer.Key("serviceProperties"); 
	writer.StartObject(); 

	//dataProperties 
	writer.Key("dataProperties"); 
	writer.StartObject(); 

	int  dataTime = 0;
	bool bb = false;
	bool IsHavePowerEvent = false;
	int  iEventVal = -1;
	for( int i=0; i<rtData.size(); i++ )
	{
		RT_DATA & rt = rtData[i]; 

		if( rt.update_time <= 0 ) continue;
			
		sprintf(tmp, "%s", rt.field_name );
		writer.Key(tmp);  
		
		if( rt.data_type  == DATA_CATEGORY_YC ) {
			
			if( strstr(pServiceID, "Accumulator") != NULL ) {
				sprintf(tmp, "%0.5f", rt.fval );
			}
			else {
				sprintf(tmp, "%0.2f", rt.fval );
			}
			writer.String(tmp);  
		}
		else {
			sprintf(tmp, "%d", (int)rt.fval );
			writer.String(tmp); 
		}

		if( strcmp(rt.field_name, "PwrOffAlm") == 0 )
		{
			IsHavePowerEvent = true;
			iEventVal = (int)rt.fval;
			dataTime = rt.update_time;
		}
		else 
		{
			if( dataTime <= 0 ) dataTime = rt.update_time;
		}
		


		bb = true;
	}

	writer.EndObject();

	//dataTime 
	writer.Key("dataTime"); 
	writer.String(base::toTimeString(dataTime).c_str());  


	writer.EndObject();
	writer.EndObject();
	writer.EndObject(); 

	if( !bb ) {
		base::my_printf("send payload: devtype=%s nodeId=%s serverId=%s ,field_name valid size=0 \n", pDevInfo->deviceType,pDevInfo->dev_nodeID, pServiceID  );
		return false;
	}
		 
	pubmsg.payload = (char*)s.GetString();
	pubmsg.payloadlen = s.GetSize() ;  
	pubmsg.retained = 0;
	pubmsg.qos = qos;
				
	opts.onSuccess =  onPubSend;
	opts.context = client;  

	base::my_printf( "send payload: %s\n", s.GetString());

 	int rc = MQTTAsync_send(client, topic, pubmsg.payloadlen, pubmsg.payload,pubmsg.qos, pubmsg.retained, &opts);
	if ( rc  != MQTTASYNC_SUCCESS )
	{  
		base::my_printf( "Failed to  sendMessage  , return code %d\n", rc);
		return false;
	} 

	if( IsHavePowerEvent && load_rt_type==2 )  
	{ 
		_Put_Event(pDevInfo, dataTime, iEventVal);
		_send_event_data(client, qos,  pDevInfo,  dataTime, iEventVal);
	}
	 
	return true;
}

void _send_rt_data(struct pub_thread_parms & parms, bool bSendChangeData, unsigned int load_rt_type, std::vector<FORE_DEV_DICT> & aryDev, 
			std::map<string,std::vector<RT_DATA> > & mapNodeID2Rtdata,	std::map<string, ARYMEATEMPLATE> & mapServiceID2Meatempl )
{
	//base::my_printf("aryDev=%d mapNodeID2Rtdata.size=%d, mapServiceID2Meatempl=%d \n ", aryDev.size(), mapNodeID2Rtdata.size(), mapServiceID2Meatempl.size() );

	for( int i=0; i<aryDev.size(); i++ )
	{
		FORE_DEV_DICT & dev = aryDev[i];
		
		std::map<string, std::vector<RT_DATA> >::iterator itRtdata = mapNodeID2Rtdata.find( dev.dev_nodeID );
		if(itRtdata == mapNodeID2Rtdata.end()) { 
			//base::my_printf("nodeid=%s \n", dev.dev_nodeID);
			continue; 
		}

		
		std::vector<RT_DATA> &  rtAllData =  itRtdata->second; 
		 
		if( rtAllData.size() <= 0 ) { 
			//base::my_printf("nodeid=%s rtAllData.size=%d \n", dev.dev_nodeID, rtAllData.size());
			continue;
		}
		
		std::vector<RT_DATA>  rtDataChanged;
		if( !bSendChangeData ) {  //��ǰ����ֻ�з�ȫ����ʱ�ŵ��ô˲��֣���Ϊȫ���ݷ���ʱ��ֻ���ٴ�װ�������Ƿ�仯 
 			TDbFun::load_rtdata_bychanged( rtAllData, rtDataChanged);  
		}
		else { 
			/*bool byx = false;
			if(  loop_count == 0 )  { 
				TDbFun::load_rtdata_bychanged( rtAllData,  rtDataChanged, 1);  //ֻװ��yc���ɣ� ��Ϊyx��200�����װ��һ��
			}
			else { 
				byx = true; 
				TDbFun::load_rtdata_bychanged( rtAllData,  rtDataChanged, 2);  //load changed yx
			} */
			
			TDbFun::load_rtdata_bychanged( rtAllData,  rtDataChanged, load_rt_type);  //0:all, 1:yc, 2:yx, 3:changeYxYc
		}
		
		/*if(  loop_count != 0 && rtDataChanged.size() <0  )  {
			base::my_printf("nodeid=%s rtAllData.size=%d loop_count=%d rtDataChanged.size=%d \n", dev.dev_nodeID, rtAllData.size(), loop_count, rtDataChanged.size());
			continue;  //û�з����仯
		}*/
		if( (load_rt_type == 3 || load_rt_type == 2 ) && rtDataChanged.size() <=0  )  {
			//base::my_printf("nodeid=%s rtAllData.size=%d  rtDataChanged.size=%d \n", dev.dev_nodeID, rtAllData.size(),  rtDataChanged.size());
			continue;  //û�з����仯
		}
		
		 
		//���ڻ�Ϊ�ɰ汾ֻ�ܰ����������������ݣ�����һ���豸���뷢�Ͷ��
		for (std::map<string, ARYMEATEMPLATE>::iterator it = mapServiceID2Meatempl.begin(); it != mapServiceID2Meatempl.end();  ++it) 
		{   
			string			strServiceID			= (*it).first; 
			 
			std::vector<RT_DATA>  tmpAryRtData;
			if( load_rt_type == 0 || load_rt_type == 1) {//all_rt   or all_yc
				for (int m=0; m<rtAllData.size(); m++) { 
					
					if( load_rt_type == 1 && rtAllData[m].data_type == DATA_CATEGORY_YX ) continue;  //all_ycʱ��������yx��yx���Ǳ仯ʱ��������

					if( strcmp(strServiceID.c_str(), rtAllData[m].serverId) == 0 ) {
						tmpAryRtData.push_back( rtAllData[m] ); 
					}
				}	
			} 
			else {  
				for (int m=0; m<rtDataChanged.size(); m++) { 
					if( strcmp(strServiceID.c_str(), rtDataChanged[m].serverId) == 0 ) {
						tmpAryRtData.push_back( rtDataChanged[m] ); 
					}
				}	
			}

			if( tmpAryRtData.size() > 0 ) {  
	
				_send_sub_data(parms.c, parms.qos, &dev, (char*)strServiceID.c_str(), tmpAryRtData, load_rt_type)  ; 
			} 
		} 
	} 
	
}
 

thread_return_type process_pubMsg(void* n)
{	
	int i = 0;
	int rc = 0;
	int wait_seconds = 30; 
   

	struct pub_thread_parms * pParm = (pub_thread_parms*)n;
	
	struct pub_thread_parms  parms; 
	parms.c = pParm->c;
	parms.index = pParm->index;
	parms.qos   = pParm->qos;
	parms.devTypeInfo  = pParm->devTypeInfo;
	parms.pPubCli = pParm->pPubCli;
	delete n;
	n = NULL;
	 
	base::MySleep(1000);
 
	pthread_t   cur_tid = pthread_self();
	base::my_printf( "Thread id=%lu, devtype=%s  begin.... \n", cur_tid,  parms.devTypeInfo.dev_type );

	

	pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL); //�����˳��߳� 
	pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL); //��������ȡ�� 
 

	std::map<string, std::vector<RT_DATA> >  mapNodeID2Rtdata; 

	std::map<string, ARYMEATEMPLATE>  tmp_mapServiceID2Meatempl; 
	TDbFun::query_meaTempl_by_devType(parms.devTypeInfo.dev_type, tmp_mapServiceID2Meatempl);
	if( tmp_mapServiceID2Meatempl.size() <= 0 ) {
		base::my_printf("THREAD id=%lu devtype=%s finished \n", cur_tid, parms.devTypeInfo.dev_type ); 
		parms.pPubCli->CloseThreadHandle( parms.devTypeInfo.dev_type );
		return NULL;
	}

	std::map<string, ARYMEATEMPLATE>  mapServiceID2Meatempl; 

	for (std::map<string, ARYMEATEMPLATE>::iterator it = tmp_mapServiceID2Meatempl.begin(); it != tmp_mapServiceID2Meatempl.end();  ++it) 
	{   
		string			strServiceID			= (*it).first; 
		ARYMEATEMPLATE		tmp_ary				= (*it).second;  

		ARYMEATEMPLATE   ary;
		for( int i=0; i<tmp_ary.size(); i++ )
		{
			FORE_MEA_TEMPLATE & rec = tmp_ary[i];
			if( rec.zf_type == 0 ) //�Ƿ�ת����ƽ̨ ��Ĭ��Ϊ0 ת�������ǵ���ʷ��¼���� ֵΪ1�� ��ת��
				ary.push_back( rec );
		} 

		mapServiceID2Meatempl.insert( pair<string, ARYMEATEMPLATE>(strServiceID, ary) );

	}
	  
	
	std::vector<FORE_DEV_DICT>   aryDev;
	TDbFun::GetDevs_ByDevType( parms.devTypeInfo.dev_type, aryDev ); 

	base::my_printf(" thread:  devType=%s  count=%d \n", parms.devTypeInfo.dev_type, aryDev.size() );
	
	if( aryDev.size() <= 0 ) {
		base::my_printf("THREAD id=%lu devtype=%s finished \n", cur_tid, parms.devTypeInfo.dev_type ); 
		parms.pPubCli->CloseThreadHandle( parms.devTypeInfo.dev_type );
		return NULL;
	}

	//�����豸���ͣ� ��ÿһ���豸�ķ��������� ��������
	for( int i=0; i<aryDev.size(); i++ )
	{
		FORE_DEV_DICT & dev = aryDev[i];
	 
		std::vector<RT_DATA>  rtAllData;
		for (std::map<string, ARYMEATEMPLATE>::iterator it = mapServiceID2Meatempl.begin(); it != mapServiceID2Meatempl.end();  ++it) 
		{   
				string			strServiceID			= (*it).first; 
				ARYMEATEMPLATE		ary				= (*it).second;  
				if( ary.size() <= 0 )  continue;
				  
				std::vector<RT_DATA>  tmpRtAllData;
				TDbFun::load_rtdata( dev.dev_nodeID, ary, tmpRtAllData);  
				if( tmpRtAllData.size() <= 0 ) continue;  
				 
				//����û�жԷְ����� 
				//_send_sub_data(parms.c, parms.qos, &dev, (char*)strServiceID.c_str(), tmpRtAllData)  ; 
				_send_sub_data(parms.c, parms.qos, &dev, (char*)strServiceID.c_str(), tmpRtAllData, 0)  ;  //0:all

				rtAllData.insert(rtAllData.end(), tmpRtAllData.begin(), tmpRtAllData.end()); 
				
				//base::MySleep( 100 );
		}
		
		if( rtAllData.size() > 0 ) {
			//base::my_printf(" nodeID =%s  rtAllData.size=%d \n", dev.dev_nodeID, rtAllData.size() );
			mapNodeID2Rtdata[dev.dev_nodeID] = rtAllData ; 
		}
	}
	  

	bool bSendChangeData = true;    //�仯ֵ����
    	int  allData_time_scan = 300;     //300s
	unsigned int loop_count = 0;

	if( parms.devTypeInfo.Reporting_interval > 300 ) {
		allData_time_scan = parms.devTypeInfo.Reporting_interval;
	}

	if( allData_time_scan != 0 && allData_time_scan < 5 ) 
		allData_time_scan = 300;

	unsigned int load_rt_type = 1;   //	0:all, 1:yc, 2:yx, 3,changeYxyc
	while( 1 )
	{    
  
		pthread_testcancel();
     
		if( !bSendChangeData )    //ֻ��ȫ���ݣ����ڿ��Ը��������ļ���
		{     
			load_rt_type = 0;
			_send_rt_data(parms, bSendChangeData, load_rt_type, aryDev, mapNodeID2Rtdata, mapServiceID2Meatempl); 
			
			base::MySleep(1000*allData_time_scan );  //allData_time_scan  s   
		}
		else
		{   
			if( allData_time_scan == 0 ) 
			{
				if( loop_count % (60*5)  == 0 ) {  //all
					load_rt_type = 0;
					base::my_printf("load_rt_type=%d \n", load_rt_type); 
					_send_rt_data(parms, bSendChangeData, load_rt_type, aryDev, mapNodeID2Rtdata, mapServiceID2Meatempl);
				}
				else {
					load_rt_type = 3;
					_send_rt_data(parms, bSendChangeData, load_rt_type, aryDev, mapNodeID2Rtdata, mapServiceID2Meatempl);   
				}
			}
			else 
			{ 
				if( (++loop_count) % (allData_time_scan*5) == 0 )  
				{
					//if( loop_count % (60*5)  == 0 ) {
						load_rt_type = 0;
					//}
					//else
					//	load_rt_type = 1;
	
					base::my_printf("load_rt_type=%d \n", load_rt_type);

					_send_rt_data(parms, bSendChangeData, load_rt_type, aryDev, mapNodeID2Rtdata, mapServiceID2Meatempl);
				}	 
				else
				{ 	//base::my_printf("loop_count=%d \n", loop_count);
					//if( loop_count == 0 ) loop_count++;  //��ֹloop_countѭ����0���ͻᶪʧһ֡�仯���ݡ�
					load_rt_type = 2;
					_send_rt_data(parms, bSendChangeData, load_rt_type, aryDev, mapNodeID2Rtdata, mapServiceID2Meatempl);  
	 			} 
			}

			if( loop_count % (60*2)  == 0 ) 
			{
				std::vector<FORE_DEV_DICT>   aryTmpDev;
				TDbFun::GetDevs_ByDevType( parms.devTypeInfo.dev_type, aryTmpDev ); 
				if(aryTmpDev.size() != aryDev.size() )
				  break;
			}

			base::MySleep(200);   
		} 

			 
	} 
 
	parms.pPubCli->CloseThreadHandle(parms.devTypeInfo.dev_type); 

	base::my_printf("THREAD devType=%s finished \n", parms.index, parms.devTypeInfo.dev_type );
 	  
	return NULL;
 
}


bool CMQTTAsyncPubCli::on_subscribe()
{

	return true;
}



int Pub_msgarrvd(void* context, char* topicName, int topicLen, MQTTAsync_message* m)
{
	CMQTTAsyncPubCli * ipCli = (CMQTTAsyncPubCli*)context; 
	 

	return 1;
}
